package com.fishleap.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author zlg
 */
public class HbaseClientDemo {

    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;

    @Before
    public void init() throws IOException {
        // 获取一个配置文件对象
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "atzlg6");
//        conf.set("hbase.zookeeper.property.clientPort", "2181");
        // 通过配置对象获取连接
        conn = ConnectionFactory.createConnection(conf);
    }

    // 创建一张hbase表
    @Test
    public void createTable() throws IOException {
        admin = (HBaseAdmin) conn.getAdmin();
        //创建表描述器
        HTableDescriptor teacher = new HTableDescriptor(TableName.valueOf("relationship"));
        //设置列族描述器
        teacher.addFamily(new HColumnDescriptor("friends"));
        //执行创建操作
        admin.createTable(teacher);
        System.out.println("relationship表创建成功！！");
    }

    //插入多条数据
    @Test
    public void putData() throws IOException {
        //获取一个表对象
        Table t = conn.getTable(TableName.valueOf("relationship"));
        List<Put> putList = new ArrayList<Put>();
        //设定rowkey
        Put put = new Put(Bytes.toBytes("jack"));
        //列族，列名，列值value
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("mary"), Bytes.toBytes("mary"));
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("tom"), Bytes.toBytes("tom"));
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("cita"), Bytes.toBytes("cita"));
        putList.add(put);
        //批量执行插入数据
        t.put(putList);

        Put put2 = new Put(Bytes.toBytes("mary"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("jack"), Bytes.toBytes("jack"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("tom"), Bytes.toBytes("tom"));
        t.put(put2);
        Put put3 = new Put(Bytes.toBytes("tom"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("jack"), Bytes.toBytes("jack"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("cita"), Bytes.toBytes("cita"));
        t.put(put3);
        Put put4 = new Put(Bytes.toBytes("cita"));
        put4.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("jack"), Bytes.toBytes("jack"));
        put4.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("mary"), Bytes.toBytes("mary"));
        t.put(put4);
        //关闭table对象
        t.close();
        System.out.println("插入成功！！");
    }

    //删除一条数据
    @Test
    public void deleteData() throws IOException {
        //需要获取一个table对象
        final Table worker = conn.getTable(TableName.valueOf("relationship"));
        //准备delete对象:rowkey
        final Delete delete = new Delete(Bytes.toBytes("cita"));
        //执行删除
        worker.delete(delete);
        //关闭table对象
        worker.close();
        System.out.println("删除数据成功！！");
    }

    //查询某个列族数据
    @Test
    public void getDataByCF() throws IOException {
        //获取表对象
        HTable student = (HTable) conn.getTable(TableName.valueOf("relationship"));
        //创建查询的get对象
        Get get = new Get(Bytes.toBytes("jack"));
        //指定列族信息
        // get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"));
        get.addFamily(Bytes.toBytes("friends"));
        //执行查询
        Result res = student.get(get);
        Cell[] cells = res.rawCells();//获取改行的所有cell对象
        for (Cell cell : cells) {
            //通过cell获取rowkey,cf,column,value
            String cf = Bytes.toString(CellUtil.cloneFamily(cell));
            String column = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            System.out.println(rowkey + "----" + cf + "---" + column + "---" + value);
        }
        student.close();//关闭表对象资源
    }

    //通过Scan全表扫描
    @Test
    public void scanAllData() throws IOException {
        HTable student = (HTable) conn.getTable(TableName.valueOf("relationship"));
        Scan scan = new Scan();
        ResultScanner resultScanner = student.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取改行的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowkey,cf,column,value
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
            }
        }
        student.close();
    }

    //通过startRowKey和endRowKey进行扫描查询：指定scan的开始rowkey和结束rowkey
    @Test
    public void scanRowKey() throws IOException {
        HTable teacher = (HTable) conn.getTable(TableName.valueOf("relationship"));
        Scan scan = new Scan();
        //指定查询的rowkey区间，rowkey在hbase中以字典序排列
        scan.setStartRow("0001".getBytes());
        scan.setStopRow("2".getBytes());
        ResultScanner resultScanner = teacher.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取改行的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowkey,cf,column,value
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
            }
        }
        teacher.close();
    }

    // 释放资源
    @After
    public void destroy() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        // 释放连接
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}